package 生产者;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * acks=0：生产者不会等待任何来自服务器的确认。这意味着如果发送失败，我们无法知道。这种方式提供了最低的延迟但最高的风险。
 * acks=1：生产者会收到leader副本的确认。只要leader接收到消息就会返回确认，不等待其他follower副本同步该消息。这种设置提供了一个平衡点，在可靠性和性能之间取得妥协。
 * acks=all：这是最严格的确认机制，要求所有当前同步(in-sync)的副本都确认收到了消息。这种方式保证了即使leader副本失败，也有完整的备份，但会带来最大的延迟。
 *
 *
 * /**
 *  * 在Kafka中，ISR代表"In-Sync Replicas"，即同步副本集合。它是一个非常重要的概念，用于确保消息的高可靠性和可用性。ISR指的是对于某个特定的主题分区(partition)，当前与leader保持同步的所有副本(replica)的集合。这意味着这些副本不仅包括leader本身，还包括那些能够及时跟上leader更新速度的follower副本。
 *  *
 *  * ISR的工作原理
 *  * Leader和Follower：每个分区都有一个leader副本和零个或多个follower副本。Leader负责处理所有针对该分区的读写请求，而follower则被动地从leader复制数据。
 *  * 同步状态：当一个follower能够跟上leader的进度，也就是说它没有落后太多（由replica.lag.time.max.ms配置控制），它就被认为是“in-sync”的，即属于ISR集合的一部分。如果follower由于某些原因（如网络延迟、负载过重等）落后于leader超过设定的时间阈值，它将被暂时从ISR中移除，直到再次赶上leader的进度。
 *  * 可靠性保证：通过调整acks参数，生产者可以指定消息发送成功需要多少个副本确认。例如，设置acks=all意味着只有当ISR中的所有副本都确认收到消息后，这条消息才被认为是成功提交的。这样即使leader突然失败，新的leader也能从ISR中选出，并且不会丢失任何已经确认的消息。
 *  * 重要性
 *  * ISR机制提供了对故障恢复和数据一致性之间的平衡。虽然增加ISR中的副本数量可以提高系统的容错能力，但它也可能影响性能，因为更多的副本需要同步数据。因此，在实际部署时，需要根据业务需求来权衡这两方面因素。
 *  *
 *  * 理解ISR有助于更好地配置Kafka集群以满足不同场景下的可用性和持久性要求。例如，在对数据一致性和耐久性要求较高的系统中，你可能希望ISR包含尽可能多的副本；而在更注重吞吐量和延迟的应用里，则可以选择较少但足够可靠的副本数目。
 *  *
 *  *
 *  */
public class KafkaAcksExample {

    public static void main(String[] args) {
        // 设置不同的acks值来演示其区别
        String[] acksValues = {"0", "1", "all"};

        for (String acks : acksValues) {
            Properties props = new Properties();
            // 指定Kafka集群地址
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "114.251.235.19:9092");
            // 设置acks配置
            props.put(ProducerConfig.ACKS_CONFIG, acks);
            // 消息序列化器
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

            KafkaProducer<String, String> producer = new KafkaProducer<>(props);

            try {
                // 创建要发送的消息记录
                ProducerRecord<String, String> record = new ProducerRecord<>("topic_test", "key", "value");

                // 发送消息
                producer.send(record).get(); // 使用get()方法进行同步发送，便于示例展示

                System.out.println("Message sent with acks=" + acks);
            } catch (Exception e) {
                System.out.println("Failed to send message with acks=" + acks);
                e.printStackTrace();
            } finally {
                // 关闭生产者
                producer.close();
            }
        }
    }
}